package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseAppV2;
import com.atguigu.gmall.realtime.bean.OrderDetail;
import com.atguigu.gmall.realtime.bean.OrderInfo;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.function.DimAsyncFunction;
import com.atguigu.gmall.realtime.util.DimUtil;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.text.ParseException;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static com.atguigu.gmall.realtime.common.Constant.TOPIC_DWD_ORDER_DETAIL;
import static com.atguigu.gmall.realtime.common.Constant.TOPIC_DWD_ORDER_INFO;

/**
 * 订单宽表
 * @author Archie
 * @date 2021-10-21 14:53
 * @description
 */
public class DwmOrderWide_Cache_Async extends BaseAppV2 {

	public static void main(String[] args) {
		new DwmOrderWide_Cache_Async().init(
				3003,
				1,
				"DwmOrderWide_Cache_Async",
				"DwmOrderWide_Cache_Async",
				TOPIC_DWD_ORDER_INFO, TOPIC_DWD_ORDER_DETAIL);
	}

	@Override
	protected void run(StreamExecutionEnvironment env,
					   Map<String, DataStreamSource<String>> streams) {

		// 1. 双流JOIN
		SingleOutputStreamOperator<OrderWide> orderWideSingleOutputStream = joinFact(streams);
		// 2. 补充维度信息（redis缓存 + 异步发送）
		SingleOutputStreamOperator<OrderWide> orderWideOutputStream = joinDim(orderWideSingleOutputStream);
		// 3. 写出到Kafka
		writeToKafka(orderWideOutputStream);

	}

	private void writeToKafka(SingleOutputStreamOperator<OrderWide> orderWideOutputStream) {
		orderWideOutputStream
				.map(JSON::toJSONString)
				.addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWM_ORDER_WIDE));
	}

	private SingleOutputStreamOperator<OrderWide> joinDim(SingleOutputStreamOperator<OrderWide> stream) {
		return AsyncDataStream.unorderedWait(
				stream,
				new DimAsyncFunction<OrderWide>() {
					@Override
					public void addDim(Connection conn, Jedis jedisConn, OrderWide orderWide, ResultFuture<OrderWide> resultFuture) {
						// 1. 补齐的用户维度
						JSONObject userInfo = DimUtil.getDim(jedisConn, conn, "dim_user_info", orderWide.getUser_id());
						orderWide.setUser_gender(userInfo.getString("GENDER"));
						try {
							orderWide.calcUser_Age(userInfo.getString("BIRTHDAY"));
						} catch (ParseException e) {
							e.printStackTrace();
						}
						// 2. 补齐省份信息
						JSONObject baseProvince = DimUtil.getDim(jedisConn, conn, "dim_base_province", orderWide.getProvince_id());
						orderWide.setProvince_name(baseProvince.getString("NAME"));
						orderWide.setProvince_iso_code(baseProvince.getString("ISO_CODE"));
						orderWide.setProvince_area_code(baseProvince.getString("AREA_CODE"));
						orderWide.setProvince_3166_2_code(baseProvince.getString("ISO_3166_2"));
						// 3. 补齐skuInf
						JSONObject skuInfo = DimUtil.getDim(jedisConn, conn, "dim_sku_info", orderWide.getSku_id());
						orderWide.setSku_name(skuInfo.getString("SKU_NAME"));
						/* 补齐 spu_id  \ tm_id \ c3_id */
						orderWide.setSpu_id(skuInfo.getLong("SPU_ID"));
						orderWide.setTm_id(skuInfo.getLong("TM_ID"));
						orderWide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
						// 4. spu
						JSONObject spuInfo = DimUtil.getDim(jedisConn, conn, "dim_spu_info", orderWide.getSpu_id());
						orderWide.setSpu_name(spuInfo.getString("SPU_NAME"));
						// 5. tm
						JSONObject tmInfo = DimUtil.getDim(jedisConn, conn, "dim_base_trademark", orderWide.getTm_id());
						orderWide.setTm_name(tmInfo.getString("TM_NAME"));
						// 6. c3
						JSONObject c3Info = DimUtil.getDim(jedisConn, conn, "dim_base_category3", orderWide.getCategory3_id());
						orderWide.setCategory3_name(c3Info.getString("NAME"));
						// 封装数据到future中
						resultFuture.complete(Collections.singletonList(orderWide));
					}
				},
				60,
				TimeUnit.SECONDS
		);
	}

	private SingleOutputStreamOperator<OrderWide> joinFact(Map<String, DataStreamSource<String>> streams) {
		KeyedStream<OrderInfo, Long> orderInfoStream = streams
				.get(TOPIC_DWD_ORDER_INFO)
				.map(data -> JSON.parseObject(data, OrderInfo.class))
				.assignTimestampsAndWatermarks(
						WatermarkStrategy
								.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
								.withTimestampAssigner((info, ts) -> info.getCreate_ts())
				)
				.keyBy(OrderInfo::getId);

		KeyedStream<OrderDetail, Long> orderDetailStream = streams
				.get(TOPIC_DWD_ORDER_DETAIL)
				.map(data -> JSON.parseObject(data, OrderDetail.class))
				.assignTimestampsAndWatermarks(
						WatermarkStrategy
								.<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
								.withTimestampAssigner((detail, ts) -> detail.getCreate_ts())
				)
				.keyBy(OrderDetail::getOrder_id);

		return orderInfoStream
				.intervalJoin(orderDetailStream)
				.between(Time.seconds(-10), Time.seconds(10))
				.process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
					@Override
					public void processElement(OrderInfo left,
											   OrderDetail right,
											   Context ctx,
											   Collector<OrderWide> out) throws Exception {
						out.collect(new OrderWide(left, right));
					}
				});
	}

}
